package com.zyx.flinkdemo.dataset.source;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.types.Row;
import ru.yandex.clickhouse.ClickHouseDriver;

/**
 * @author Yaxi.Zhang
 * @since 2021/8/19 10:45
 * desc: 通过JDBCInputFormat查询Clickhouse代码
 */
public class JdbcInputFormatDemo {

    public static final String USER_NAME = "";
    public static final String PASSWORD = "";
    public static final String DB_URL = "jdbc:clickhouse://bigdata022:8123";
    public static final int FETCH_SIZE = 50;

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        JdbcInputFormat jdbcInputFormat = JdbcInputFormat.buildJdbcInputFormat()
                .setDrivername(ClickHouseDriver.class.getName())
                .setDBUrl(DB_URL)
                .setUsername(USER_NAME)
                .setPassword(PASSWORD)
                .setFetchSize(FETCH_SIZE)
                .setQuery("select assetId, curPrice from dbtest.ticker_chip limit 1000")
                .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.DOUBLE_TYPE_INFO))
                .finish();

        DataSource<Row> rowDataSource = env.createInput(jdbcInputFormat);

        rowDataSource.print();

        //env.execute();
    }
}
